package com.qisi.functions;

import com.qisi.demo.SenderService;
import com.qisi.pojo.AppType;
import com.qisi.pojo.KafkaAlertLog;
import com.qisi.pojo.alertManager.Route;
import com.qisi.services.AlertManagerService;
import com.qisi.services.ISendService;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.io.IOException;


/**
 * 发送 KafkaAlertLo.genAlertManageData() 到 AlertManager
 */
@Slf4j
public class AlertManagerSinkFunction extends RichSinkFunction<KafkaAlertLog> {

    private ISendService alertManagerService;

    private AppType route;

    public AlertManagerSinkFunction(Configuration config) throws IOException {

        ConfigOption<AppType> routeOption = ConfigOptions.key("app.type").enumType(AppType.class).noDefaultValue();
        ConfigOption<String> hostOption = ConfigOptions.key("am.host").stringType().defaultValue("10.6.215.39");
        ConfigOption<Integer> portOption = ConfigOptions.key("am.port").intType().defaultValue(9093);

        route = config.get(routeOption);
        // alertManagerService = new AlertManagerService(config.get(hostOption), config.get(portOption));
        alertManagerService = new SenderService(config.get(hostOption), config.get(portOption));
    }

    @Override
    public void invoke(KafkaAlertLog value, Context context) {
        switch (this.route){

            case CLIENT_EVENT:  // 车机端埋点
                alertManagerService.send(value.genClientAlertData()); break;

            case CLOUD_LOG:   // 云端微服务日志告
                alertManagerService.send(value.genCloudServiceLogAlertData()); break;

            case CLOUD_EVENT: // 云端埋点日志
                alertManagerService.send(value.genCloudEventAlertData()); break;

            case SENTRY:  // 车机端埋点
                alertManagerService.send(value.genSentryAlertData()); break;

            default: {
                log.error("unknown route: {}, route can not null ", this.route);
                throw new RuntimeException("unknown route, route can not null ");
            }

        }
    }

    @Override
    public void close() throws Exception {
        alertManagerService.close();
    }

}
